package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.OrderPromoService;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.order.dto.CreateSeckillOrderResponse;
import com.mall.promo.constant.PromoRetCode;
import org.apache.dubbo.config.annotation.Reference;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/*
 * @description: MQ事务消息，生成订单
 */
@Component
public class CreatePromoOrderConsumer {

    private DefaultMQPushConsumer consumer;

        @Reference(check = false, timeout = 3000, retries = 0)
        private OrderPromoService orderPromoService;

        @PostConstruct
        public void init() throws MQClientException {

        //初始化消费者
        consumer = new DefaultMQPushConsumer("promo_order_consumer");
        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.subscribe("create_promo_order", "*");

        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                MessageExt message = list.get(0);
                //将获得的字节数组转化成字符串-->CreateSeckillOrderRequest对象
                byte[] body = message.getBody();
                String requestStr = new String(body, 0, body.length);
                CreateSeckillOrderRequest request = JSON.parseObject(requestStr, CreateSeckillOrderRequest.class);
                CreateSeckillOrderResponse response = orderPromoService.createPromoOrder(request);

                if (!response.getCode().equals(PromoRetCode.SUCCESS.getCode())) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

    }
}
